/*
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include "BTreeIndex.h"
#include "BTreeNode.h"
#include <queue>
#include <cstdarg>

using namespace std;

#define DEBUG_MODE false

/*
 * BTreeIndex constructor
 */
BTreeIndex::BTreeIndex()
{
	treeHeight = 0;
    rootPid = -1;

}

void BTreeIndex::debug(const char* fmt, ...)
{
    va_list args;
    va_start(args,fmt);
    if(DEBUG_MODE){
    	vprintf(fmt,args);
    }
    va_end(args);
}

RC BTreeIndex::save(){
	char buffer[PageFile::PAGE_SIZE];

	int offset = 0;

	//Clear the buffer.
	memset(&buffer[offset], 0, PageFile::PAGE_SIZE);

	//Copy the root page ID into the buffer.
	memcpy(&buffer[offset], &rootPid, sizeof(PageId));
	offset += sizeof(PageId);

	//Copy the three height into the buffer.
	memcpy(&buffer[offset], &treeHeight, sizeof(int));
	offset += sizeof(int);

	//Write the buffer into the page file.
	if(pf.write(0, buffer) != 0) {
		return RC_FILE_WRITE_FAILED;
	}

	return 0;

}


typedef struct {
	int height;
	PageId pid;
} q_node;

void BTreeIndex::print(){

	debug("\n------DUMPING TREE DATA------\n");
	debug("Tree Height: %d \t Root PID: %d \n", treeHeight, rootPid);
	debug("\n");

	if(treeHeight == 0 || rootPid == -1){
		return;
	}

	queue<q_node> q;
	q_node node;
	BTLeafNode* leafNode;
	BTNonLeafNode* nonLeafNode;

	//Initialize the first element of the queue.
	q_node root =  {1, rootPid};
	q.push(root);


	while(!q.empty()){


		node = q.front();
		q.pop();

		if(node.height == 1){
			debug("          <<ROOT>>          \n");
		}

		debug("          <<PID%d>>          \n", node.pid);


		//Handle leaf node.
		if(node.height == treeHeight){

			leafNode = new BTLeafNode();
			if(leafNode->read(node.pid, pf) != 0) {
				debug("Unable to read leaf node during printout.\n");
			}

			leafNode->print();
			delete leafNode;

		//Handle non leaf node.
		}else{

			//Read and print the contents of the non leaf node.
			nonLeafNode = new BTNonLeafNode();
			if(nonLeafNode->read(node.pid, pf) != 0) {
				debug("Unable to read non leaf node during printout.\n");
			}
			nonLeafNode->print();


			//Insert all the children of the non leaf node into the queue
			//to be processed later.
			for(int i = 0; i < nonLeafNode->getKeyCount(); i++){

				q_node new_node;
				new_node.height = node.height + 1;
				new_node.pid = nonLeafNode->getPageIdAt(i);
				q.push(new_node);

			}

			if(nonLeafNode->getNextNodePtr() != 0){
				q_node next_node;
				next_node.height = node.height + 1;
				next_node.pid = nonLeafNode->getNextNodePtr();
				q.push(next_node);

			}

			delete nonLeafNode;

		}


	}


}

/*
 * Open the index file in read or write mode.
 * Under 'w' mode, the index file should be created if it does not exist.
 * @param indexname[IN] the name of the index file
 * @param mode[IN] 'r' for read, 'w' for write
 * @return error code. 0 if no error
 */
RC BTreeIndex::open(const string& indexname, char mode)
{
	//Save the mode in order to know what to do when then index is closed.
	this->mode = mode;

	//Open the provided index file.
	if(pf.open(indexname, mode) != 0) {
		return RC_FILE_OPEN_FAILED;
	}

	//If the index file was just created, we cannot read any more
	//information as it was not previously stored. So just exit.
	if(pf.endPid() == 0){
		return 0;
	}

	//Copy the first page into a buffer to read information such as
	//tree height, and root page ID.
	char buffer[PageFile::PAGE_SIZE];
	if(pf.read(0, buffer) != 0) {
		return RC_FILE_READ_FAILED;
	}

	//Read root page id.
	int offset = 0;
	memcpy(&rootPid, &buffer[offset], sizeof(PageId));
	offset += sizeof(PageId);

	//Read tree height.
	memcpy(&treeHeight, &buffer[offset], sizeof(int));
	offset += sizeof(int);

	return 0;
}

/*
 * Close the index file.
 * @return error code. 0 if no error
 */
RC BTreeIndex::close()
{
	if(this->mode == 'w'){
		if(save() != 0){
			return RC_FILE_WRITE_FAILED;
		}
	}

    return pf.close();
}

/*
 * Insert (key, RecordId) pair to the index.
 * @param key[IN] the key for the value inserted into the index
 * @param rid[IN] the RecordId for the record being inserted into the index
 * @return error code. 0 if no error
 */
RC BTreeIndex::insert(int key, const RecordId& rid)
{

	debug("(#) Inserting Key: %d\n", key);

	BTLeafNode leaf;
	BTNonLeafNode node;

	//Initialize the very first root, if it hasn't been already.
	if(rootPid == -1){

		//Set the initial leaf root node information.
		treeHeight = 1;
		rootPid    = 1;

		//Save the new metadata such as tree height and root page ID in
		//page 0, in order to be able to write to page 1.
		save();

		//Write the leaf root node to the page.
		if(leaf.write(1, pf) != 0){
			debug("Unable to write leaf root.\n");
			return RC_FILE_WRITE_FAILED;

		}

	}

	//Points to a potential bucket.
	IndexCursor cursor;
	stack<PageId> path;

	locateWithPath(key, cursor, &path);

	debug("(#) Bucket at (PID: %d, EID: %d).\n", cursor.pid, cursor.eid);
	debug("(#) Path Length: %d.\n", (int)path.size());
	debug("(#) Path: ");
	stack<PageId> path2(path);
	while(!path2.empty()){
		PageId id = path2.top();
		debug("%d ", id);
		path2.pop();
	}
	debug("\n");

	if(leaf.read(cursor.pid, pf) != 0) {
		return RC_FILE_READ_FAILED;
	}

	if(!leaf.isFull()){

		leaf.insert(key, rid);
		if(leaf.write(cursor.pid, pf) != 0){
			debug("Unable to write leaf node after insert.\n");
			return RC_FILE_WRITE_FAILED;
		}

	}else{


		//The sibling created after the split.
		BTLeafNode sibling;

		//The page where the sibling will be stored.
		PageId siblingPid = pf.endPid();

		//The first key of the new sibling.
		int midKey;

		leaf.insertAndSplit(key, rid, sibling, midKey);

		debug("(#) Splitting Leaf Node (PID: %d): Sibling PID: %d, Mid Key: %d.\n", (int)cursor.pid, (int)siblingPid, midKey);

		//Write the sibling leaf node into a new page.
		if(sibling.write(siblingPid, pf) != 0){
			debug("Unable to write sibling node into a new page.\n");
			return RC_FILE_WRITE_FAILED;
		}

		leaf.setNextNodePtr(siblingPid);


		if(leaf.write(cursor.pid, pf) != 0){
			debug("Unable to update leaf node.\n");
			return RC_FILE_WRITE_FAILED;
		}

		return percolate(midKey, cursor.pid, siblingPid, path);

	}

    return 0;
}

RC BTreeIndex::percolate(int midKey, PageId& leftSiblingPid, PageId rightSiblingPid, stack<PageId> &path){

	//Need to initialize a new root.
	if(path.empty()){

		PageId newRootPid = pf.endPid();
		BTNonLeafNode root;

		debug("(#) Init New Root (PID: %d). \n", (int)newRootPid);


		root.initializeRoot(leftSiblingPid, midKey, rightSiblingPid);

		//Write the sibling leaf node into a new page.
		if(root.write(newRootPid, pf) != 0){
			debug("Unable to new root into a new page.\n");
			return RC_FILE_WRITE_FAILED;
		}

		rootPid = newRootPid;
		treeHeight ++;
		debug("(#) New Tree Height: %d. \n", treeHeight);
		return save();


	}else{

		PageId nodePid = path.top();
		path.pop();

		BTNonLeafNode node;

		if(node.read(nodePid, pf) != 0) {
			debug("Unable to read node (PID: %d) in a path.\n", nodePid);
			return RC_FILE_READ_FAILED;
		}


		//Node still has space (no overflow).
		if(!node.isFull()){

			if(node.insert(midKey, leftSiblingPid) != 0){

			}

			if(node.getMaxKey() == midKey){
				node.setNextNodePtr(rightSiblingPid);
			}

			if(node.write(nodePid, pf) != 0){
				debug("Unable to update a non leaf node in a path.\n");
				return RC_FILE_WRITE_FAILED;
			}

			return save();

		}else{

			//The sibling created after the split.
			BTNonLeafNode sibling;

			//The page where the sibling will be stored.
			PageId siblingPid = pf.endPid();

			//The first key of the new sibling.
			int newMidKey;

			node.insertAndSplit(midKey, leftSiblingPid, sibling, newMidKey);

			debug("(#) Splitting Non Leaf Node(PID: %d): Sibling PID: %d, Mid Key: %d.\n", (int) nodePid,(int)siblingPid, newMidKey);

			if(sibling.getMaxKey() == midKey){
				sibling.setNextNodePtr(rightSiblingPid);
			}

			//Write the sibling leaf node into a new page.
			if(sibling.write(siblingPid, pf) != 0){
				debug("Unable to write sibling node in a path into a new page.\n");
				return RC_FILE_WRITE_FAILED;
			}

			if(node.write(nodePid, pf) != 0){
				debug("Unable to write left node in a path into a new page.\n");
				return RC_FILE_WRITE_FAILED;
			}



			return percolate(newMidKey, nodePid, siblingPid, path);


		}


	}






}

RC BTreeIndex::locateWithPath(int searchKey, IndexCursor& cursor, stack<PageId>* path){

	if(path != NULL){
		path->empty();
	}

	//RC value returned by this function.
	int retval = 0;

	//The tree is empty.
	if(treeHeight == 0) {

		cursor.pid = 1; //Page ID.
		cursor.eid = 0; //Entry ID.

	//Root is a leaf node.
	} else if(treeHeight == 1) {

		BTLeafNode root;

		int eid;

		if(root.read(rootPid, pf) != 0) {
			return RC_FILE_READ_FAILED;
		}

		if(root.locate(searchKey, eid) != 0) {
			return RC_NO_SUCH_RECORD;
		}

		cursor.pid = rootPid;
		cursor.eid = eid;


	//Tree has at least two levels.
	} else {

		int level = 1;
		PageId nextPid = rootPid;
		BTNonLeafNode node;

		do{

			if(path != NULL){
				//Push a page that was visited during the path to
				//the leaf node. This also includes the root node.
				path->push(nextPid);
			}

			//Read in the non leaf node.
			if(node.read(nextPid, pf) != 0) {
				return RC_FILE_READ_FAILED;
			}

			//Given the search key, find the child-node pointer to follow and
			//output in nextPid.
			if(node.locateChildPtr(searchKey, nextPid) != 0) {
				return RC_NO_SUCH_RECORD;
			}

		//Continue until we reach a leaf node.
		}while(++level < treeHeight);

		cursor.pid = nextPid;
		cursor.eid = 0;

		BTLeafNode leaf;

		int eid;

		if(leaf.read(cursor.pid, pf) == 0 && leaf.locate(searchKey, eid) == 0) {
			cursor.eid = eid;
		}



	}

	return 0;

}

/*
 * Find the leaf-node index entry whose key value is larger than or
 * equal to searchKey, and output the location of the entry in IndexCursor.
 * IndexCursor is a "pointer" to a B+tree leaf-node entry consisting of
 * the PageId of the node and the SlotID of the index entry.
 *
 * Note that, for range queries, we need to scan the B+tree leaf nodes.
 * For example, if the query is "key > 1000", we should scan the leaf
 * nodes starting with the key value 1000. For this reason,
 * it is better to return the location of the leaf node entry
 * for a given searchKey, instead of returning the RecordId
 * associated with the searchKey directly.
 *
 * Once the location of the index entry is identified and returned
 * from this function, you should call readForward() to retrieve the
 * actual (key, rid) pair from the index.
 *
 * @param key[IN] the key to find.
 * @param cursor[OUT] the cursor pointing to the first index entry
 *                    with the key value.
 * @return error code. 0 if no error.
 */
RC BTreeIndex::locate(int searchKey, IndexCursor& cursor)
{
	return locateWithPath(searchKey, cursor);
}

/*
 * Read the (key, rid) pair at the location specified by the index cursor,
 * and move forward the cursor to the next entry.
 * @param cursor[IN/OUT] the cursor pointing to an leaf-node index entry in the b+tree
 * @param key[OUT] the key stored at the index cursor location.
 * @param rid[OUT] the RecordId stored at the index cursor location.
 * @return error code. 0 if no error
 */
RC BTreeIndex::readForward(IndexCursor& cursor, int& key, RecordId& rid)
{

	BTLeafNode node;

	if(node.read(cursor.pid, pf) != 0) {
		return RC_FILE_READ_FAILED;
	}

	if(node.readEntry(cursor.eid, key, rid) != 0) {
		return RC_INVALID_RID;
	}

	if (cursor.eid < node.getKeyCount() - 1) {
		cursor.eid++;

	} else {

		cursor.eid = 0;

		int next = node.getNextNodePtr();

		if(next == 0) {
			return RC_END_OF_TREE;
		}

		cursor.pid = next;
	}

    return 0;
}
